package com.hrt.datasender.mq;

import com.btcode.common.MyLog;
import com.btcode.log.ILog;
import com.hrt.datasender.datacollect.drive.modbus.tcp.NetBaseDrive;
import com.hrt.datasender.datadao.model.BcDevicesModel;
import com.hrt.datasender.datadao.model.BcSerialModel;
import com.hrt.datasender.mq.model.*;

/**
 * 消息中间件，用于调度模块、收发模块、数采模块以及UI间的通讯 Created by Chance on 2017/11/1.
 */
public class MqKit {
    private static ILog log;
    /**
     * 中心站消息队列
     *
     */
    static final String MQ_CENTER_STATION = "MQ_CENTER_STATION";
    /**
     * 数据消息队列
     */
    static final String MQ_BIZ_DATA = "MQ_BIZ_DATA";
    /**
     * 反控消息队列
     */
    static final String MQ_RECRIMINATE = "MQ_RECRIMINATE";
    /**
     * 心跳队列
     */
    static final String MQ_HEATBEAT = "MQ_HEATBEAT";
    /**
     * 补发数据队列
     */
    static final String MQ_REISSUE_DATA = "MQ_REISSUE_DATA";
    /**
     * ONENET队列
     */
    final static String MQ_ONENET = "MQ_ONENET";

    public static void productMsgToCenterStationMQ(PacketModel packetModel) {
        log.info(">>>中心站队列：" + packetModel);
        MsgQueueManager.getInstance().produce(MQ_CENTER_STATION, packetModel);
    }

    public static void consumeMsgFromCenterStationMQ(MsgListener msgListener) {
        try {
            MsgQueueManager.getInstance().consume(MQ_CENTER_STATION, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel.getMqModelType() == MqModel.MODEL_TYPE_PACKET) {
                        log.info("中心站队列>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("中心站队列消费失败", e);
        }
    }

    public static void productMsgToBizDataMQ(PacketModel packetModel) {
        log.info(">>>业务数据队列：" + packetModel);
        MsgQueueManager.getInstance().produce(MQ_BIZ_DATA, packetModel);
    }

    public static void consumeMsgFromBizDataMQ(MsgListener msgListener) {
        try {
            MsgQueueManager.getInstance().consume(MQ_BIZ_DATA, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel.getMqModelType() == MqModel.MODEL_TYPE_PACKET) {
                        log.info("业务数据队列>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("业务数据队列消费失败", e);
        }
    }

    public static void productMsgToHeatBeatMQ(PacketModel packetModel) {
        log.info(">>>心跳队列：" + packetModel);
        MsgQueueManager.getInstance().produce(MQ_HEATBEAT, packetModel);
    }

    public static void consumeMsgFromHeatBeatMQ(MsgListener msgListener) {
        try {
            MsgQueueManager.getInstance().consume(MQ_HEATBEAT, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    log.info("心跳队列>>>：" + mqModel);
                    msgListener.onReceive(mqModel);
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("心跳队列消费失败", e);
        }
    }

    public static void productMsgToRefreshRegisterMQ(PacketModel packetModel) {
        log.info(">>>ONENET队列：" + packetModel);
        MsgQueueManager.getInstance().produce(MQ_ONENET, packetModel);
    }


    public static void consumeMsgFromRefreshRegisterMQ(MsgListener msgListener) {
        try {
            MsgQueueManager.getInstance().consume(MQ_ONENET, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    log.info("ONENET队列>>>：" + mqModel);
                    msgListener.onReceive(mqModel);
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("ONENET队列消费失败", e);
        }
    }

    /**
     * 消费串口数据消息
     *
     * @param msgListener
     */
    public static void consumeMsgFromSerialMQ(BcDevicesModel deviceModel, MsgListener msgListener) {

        String queueName = MqManager.getSerialDataQueueName(deviceModel);

        try {
            MsgQueueManager.getInstance().consume(queueName, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel instanceof SerialModel) {
                        log.info("串口数据队列[" + queueName + "]>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("串口数据队列消费失败", e);
        }
    }

    /**
     * 发布串口数据消息 队列名称由 <strong>SERIAL_串口号_从站编号</strong> 组成
     *
     * @param data
     */
    public static void productMsgToSerialMQ(SerialModel data) {
        String queueName = MqManager.getSerialDataQueueName(data);

        log.info(">>>串口数据队列[" + queueName + "]：" + data);

        MsgQueueManager.getInstance().produce(queueName, data);
    }

    /**
     * 消费网口数据消息
     *
     * @param msgListener
     */
    public static void consumeMsgFromNetExportMQ(NetBaseDrive deviceModel, MsgListener msgListener) {

        String queueName = MqManager.getNetExportDataQueueName(deviceModel);

        try {
            MsgQueueManager.getInstance().consume(queueName, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel instanceof NetExportDataModel) {
                        log.info("网口数据队列[" + queueName + "]>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("网口数据队列消费失败", e);
        }
    }

    /**
     * 发布网口数据消息 队列名称由 <strong>SERIAL_串口号_从站编号</strong> 组成
     *
     * @param data
     */
    public static void productMsgToNetExportMQ(NetExportDataModel data) {
        String queueName = MqManager.getNetExportDataQueueName(data);

        log.info(">>>网口数据队列[" + queueName + "]：" + data);

        MsgQueueManager.getInstance().produce(queueName, data);
    }

    /**
     * 消费询问串口数据消息
     *
     * @param msgListener
     */
    public static void consumeMsgFromSerialAskDataMQ(BcSerialModel serialModel, MsgListener msgListener) {

        String queueName = MqManager.getSerialAskDataQueueName(serialModel);

        try {
            MsgQueueManager.getInstance().consume(queueName, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel instanceof SerialModel) {
                        log.info("询问串口数据消息队列>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("询问串口数据消息队列消费失败", e);
        }
    }

    /**
     * 发布询问串口数据消息 队列名称由 <strong>SERIAL_串口号_从站编号</strong> 组成
     *
     * @param data
     */
    public static void productAskDataMsgToSerialMQ(SerialModel data) {
        log.info(">>>询问串口数据消息队列：" + data);

        String queueName = MqManager.getSerialAskDataQueueName(data);

        MsgQueueManager.getInstance().produce(queueName, data);
    }

    /**
     * 消费OPC数据消息
     *
     * @param msgListener
     */
    public static void consumeMsgFromOpcMQ(BcDevicesModel deviceModel, MsgListener msgListener) {

        String queueName = MqManager.getOpcDataQueueName(deviceModel);

        try {
            MsgQueueManager.getInstance().consume(queueName, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel instanceof OpcModel) {
                        log.info("Opc数据队列[" + queueName + "]>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("Opc数据队列消费失败", e);
        }
    }

    /**
     * 发布OPC数据消息 队列名称由 <strong>SERIAL_串口号_从站编号</strong> 组成
     *
     * @param data
     */
    public static void productMsgToOpcMQ(OpcModel data) {
        String queueName = MqManager.getOpcDataQueueName(data);

        log.info(">>>Opc数据队列[" + queueName + "]：" + data);

        MsgQueueManager.getInstance().produce(queueName, data);
    }

    public abstract static class MsgListener implements IMsgListener {

        public MsgListener() {

        }

        // public abstract void onReceive(Object msg);
    }

    public static void init() {

        log = MyLog.getInstance().getLogger(MqKit.class);
    }

    /**
     * 发布运行信息
     *
     * @param msg
     */
    public static void productOperInfo(OperInfoModel msg) {
        String queueName = "operInfo";
        MsgQueueManager.getInstance().produce(queueName, msg);
    }

    /**
     * 获取运行信息
     *
     * @return
     */
    public static OperInfoModel getOperInfo() {
        String queueName = "operInfo";
        MqModel mqModel = getOneMsg(queueName);

        if (mqModel == null) {
            return null;
        }

        OperInfoModel operInfoModel = (OperInfoModel) mqModel;
        return operInfoModel;
    }

    /**
     * 清空运行信息
     *
     * @return
     */
    public static void clearOperInfo() {
        String queueName = "operInfo";
        MsgQueueManager.getInstance().clear(queueName);
    }




    /**
     * 获取监控日志消息
     *
     * @param log
     * @return
     */
    public static LogModel getOneMsgFromMonitorLogMQ(ILog log) {

//        if (log == null) {
//            return null;
//        }

        String queueName = "operInfo";
        MqModel mqModel = getOneMsg(queueName);

        if (mqModel == null) {
            return null;
        }

        LogModel logModel = (LogModel) mqModel;
        return logModel;
    }

    private static MqModel getOneMsg(String queueName) {
        MqModel mqModel = null;
        try {
            mqModel = MsgQueueManager.getInstance().consume(queueName);
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
            log.error("getOneMsg消费失败", e);
        }
        return mqModel;
    }

    /**
     * 获取监控日志的错误消息
     *
     * @param log
     * @return
     */
    public static LogModel getOneMsgFromMonitorErrorLogMQ(ILog log) {

        if (log == null) {
            return null;
        }

        MqModel mqModel = getOneMsg(log.getName() + ".error");

        if (mqModel == null) {
            return null;
        }

        LogModel logModel = (LogModel) mqModel;
        return logModel;
    }

    public static int getMqMsgSize(String queueName) {
        return MsgQueueManager.getInstance().size(queueName);
    }


    public static void productMsgToRecriminateMQ(PacketModel packetModel) {
        log.info(">>>反控消息队列：" + packetModel);
        MsgQueueManager.getInstance().produce(MQ_RECRIMINATE, packetModel);
    }

    public static void consumeMsgFromRecriminateMQ(MsgListener msgListener) {
        try {
            MsgQueueManager.getInstance().consume(MQ_RECRIMINATE, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    log.info("反控消息队列>>>：" + mqModel);
                    msgListener.onReceive(mqModel);
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("反控消息队列消费失败", e);
        }
    }

    public static void productMsgToReissueDataMQ(PacketModel packetModel) {
        log.info(">>>补发数据队列：" + packetModel);
        MsgQueueManager.getInstance().produce(MQ_REISSUE_DATA, packetModel);
    }

    public static void consumeMsgFromReissueDataMQ(MsgListener msgListener) {
        try {
            MsgQueueManager.getInstance().consume(MQ_REISSUE_DATA, new MsgListener() {
                @Override
                public void onReceive(MqModel mqModel) {
                    if (mqModel.getMqModelType() == MqModel.MODEL_TYPE_PACKET) {
                        log.info("补发数据队列>>>：" + mqModel);
                        msgListener.onReceive(mqModel);
                    }
                }
            });
        }
        catch (QueueHasBeenListenedException | NullQueueListenerException e) {
            e.printStackTrace();
            log.error("补发数据队列消费失败", e);
        }
    }

}
